﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

namespace CommonEngine.Worker
{
    public enum DequeueMode : int
    {
        Synchronize = 0,
        Asynchronous = 1,
        Parallel = 2
    }

    /// <summary>
    /// 工作层封装---适配多线程缓冲池处理
    /// </summary>
    public class WorkerWrapper<T> : IDisposable
    {
        private object _lockObj = new object();
        private DequeueMode _mode = DequeueMode.Synchronize;

        private Action<T, CancellationToken> _queueConsumingAction;     //消费行为
        private volatile Action<WorkerWrapper<T>> ConsumerFinishedEvent;   //消费完成
        private volatile Action<WorkerWrapper<T>> WorkerFailEvent;     //生产失败        
        private volatile Action<WorkerWrapper<T>> ConsumerFailEvent;   //消费失败

        private bool isCompleted = true;   //是否完成当前消费行为
        private BlockingCollection<T> _messageCollection = null;    //缓存队列
        //--令牌取消--
        private CancellationTokenSource _cancellationTokenSource;
        private Semaphore _sema = null;
        private ParallelOptions _options = null;
        private int _maximumCount = 50;
        private int _maxDegree = 50;
        private int _millisecondsTimeout = 3000;

        /// <summary>
        /// 队列消费行为
        /// </summary>
        public Action<T, CancellationToken> QueueConsumingAction
        {
            get
            {
                lock (_lockObj)
                {
                    return _queueConsumingAction;
                }
            }
            set
            {
                lock (_lockObj)
                {
                    _queueConsumingAction = value;
                }
            }
        }
        public volatile SynchronizationContext SyncContext;
        public IRecycleBin<T> RecycleOp { get; set; }

        /// <summary>
        /// 构造
        /// </summary>
        public WorkerWrapper()
        {
            this._mode = DequeueMode.Synchronize;
            this.Init();
        }
        public WorkerWrapper(int maximumCount)
        {
            this._mode = DequeueMode.Asynchronous;
            this._maximumCount = maximumCount;
            this.Init();
        }
        public WorkerWrapper(int maxDegree, int millisecondsTimeout)
        {
            this._mode = DequeueMode.Parallel;
            this._maxDegree = maxDegree;
            this._millisecondsTimeout = millisecondsTimeout;
            this.Init();
        }

        #region---基础行为---

        private void Init()
        {
            this.RecycleOp = this.RecycleOp ?? new RecycleBin<T>();
            this._sema = new Semaphore(this._maximumCount, this._maximumCount);
            this._options = new ParallelOptions() { MaxDegreeOfParallelism = this._maxDegree };
            this._messageCollection = new BlockingCollection<T>(int.MaxValue - 147483647);
            this._cancellationTokenSource = new CancellationTokenSource();
            this.SyncContext = new SynchronizationContext();
        }

        public bool IsCanEnqueue
        {
            get
            {
                return this._messageCollection.BoundedCapacity > this._messageCollection.Count;
            }
        }

        public int BoxSpace
        {
            get
            {
                return this._messageCollection.Count;
            }
        }

        public bool HasCancelled
        {
            get;
            private set;
        }

        public void Setup()
        {
            if (!this.isCompleted)
                return;

            try
            {
                lock (this._lockObj)
                {
                    if (!this.isCompleted)
                        return;

                    this.isCompleted = false;
                    Monitor.PulseAll(this._lockObj);
                }

                this._cancellationTokenSource = new CancellationTokenSource();

                Task.Factory.StartNew(() =>
                {
                    var c_token = this._cancellationTokenSource.Token;
                    if (this._mode == DequeueMode.Synchronize)
                        CommonAct(ConsumerDequeue_Synchronize).Invoke(c_token);
                    else if (this._mode == DequeueMode.Asynchronous)
                        CommonAct(ConsumerDequeue_Asynchronous).Invoke(c_token);
                    else
                        CommonAct(ConsumerDequeue_Parallel).Invoke(c_token);
                });
            }
            catch (Exception e)
            {
                //log
                Console.Write(e.Message);
            }
        }

        public void Cancel()
        {
            if (!this._cancellationTokenSource.IsCancellationRequested)
                this._cancellationTokenSource.Cancel(false);

            if (!this._messageCollection.IsCompleted)
                this._messageCollection.CompleteAdding();
        }

        public void Dispose()
        {
            if (!this._cancellationTokenSource.IsCancellationRequested)
                this._cancellationTokenSource.Cancel(false);

            if (this._messageCollection != null && !this._messageCollection.IsAddingCompleted)
                this._messageCollection.CompleteAdding();
        }

        #endregion

        public WorkerWrapper<T> FinishWork(Action<WorkerWrapper<T>> handler)
        {
            this.ConsumerFinishedEvent = handler;
            return this;
        }
        public WorkerWrapper<T> WorkerFail(Action<WorkerWrapper<T>> handler)
        {
            this.WorkerFailEvent = handler;
            return this;
        }
        public WorkerWrapper<T> ConsumerFail(Action<WorkerWrapper<T>> handler)
        {
            this.ConsumerFailEvent = handler;
            return this;
        }

        [MTAThread]
        public void ProduceEnqueue(T message)
        {
            if (message == null)
                return;

            if (!this.IsCanEnqueue)
            {
                if (this.WorkerFailEvent != null)
                    Task.Run(() => this.WorkerFailEvent.Invoke(this));

                return;
            }

            if (this._messageCollection.IsAddingCompleted)
                return;

            if (this._messageCollection.Count < this._messageCollection.BoundedCapacity)
                this._messageCollection.Add(message);

            this.Setup();
        }

        private void ConsumerDequeue_Synchronize(CancellationToken c_token)
        {
            ConcurrentBag<T> TError = new ConcurrentBag<T>();
            foreach (var message in this._messageCollection.GetConsumingEnumerable(c_token))
            {
                try
                {
                    if (this._queueConsumingAction != null)
                        this._queueConsumingAction(message, c_token);
                }
                catch
                {
                    TError.Add(message);
                }

                Thread.Sleep(20);
                if (this._messageCollection.Count < 1)
                    break;
            }

            if (this.RecycleProblemData(TError))
                throw new ApplicationException("Deferred execution of task problem data");
        }

        private void ConsumerDequeue_Asynchronous(CancellationToken c_token)
        {
            List<Task> tLs = new List<Task>();
            ConcurrentBag<T> TError = new ConcurrentBag<T>();
            foreach (var message in this._messageCollection.GetConsumingEnumerable(c_token))
            {
                var t = Task.Run(() =>
                {
                    this._sema.WaitOne();
                    Thread.Sleep(20);

                    try
                    {
                        if (this._queueConsumingAction != null)
                            this._queueConsumingAction(message, c_token);
                    }
                    catch
                    {
                        TError.Add(message);
                    }
                    finally
                    {
                        this._sema.Release();
                    }
                });
                tLs.Add(t);

                if (!c_token.CanBeCanceled)
                    break;

                if (this._messageCollection.Count < 1)
                    break;
            }
            Task.WaitAll(tLs.ToArray(), c_token);

            if (this.RecycleProblemData(TError))
                throw new ApplicationException("Deferred execution of task problem data");
        }

        private void ConsumerDequeue_Parallel(CancellationToken c_token)
        {
            this._options.CancellationToken = c_token;
            ParallelLoopState PLS = null;

            ConcurrentBag<T> CB = new ConcurrentBag<T>(this.GetConsuming(c_token));
            ConcurrentBag<T> TError = new ConcurrentBag<T>();
            Thread.Sleep(20);
            ParallelLoopResult pr = Parallel.ForEach(CB,
                this._options, (message, loop) =>
                {
                    try
                    {
                        if (this._queueConsumingAction != null)
                            this._queueConsumingAction(message, c_token);

                        if (!c_token.CanBeCanceled)
                        {
                            PLS = loop;
                            loop.Stop();
                        }
                    }
                    catch
                    {
                        TError.Add(message);
                    }
                });

            do
            {
                Thread.Sleep(this._millisecondsTimeout);
                if (PLS != null)
                {
                    if (PLS.IsStopped)
                        break;
                    if (PLS.IsExceptional)
                        break;
                }
            }
            while (!pr.IsCompleted);

            if (this.RecycleProblemData(TError))
                throw new ApplicationException("Deferred execution of task problem data");
        }

        #region---辅助行为---

        private bool RecycleProblemData(ConcurrentBag<T> TError)
        {
            if (this.RecycleOp != null)
            {
                Thread.Sleep(20);
                if (TError.Count > 0)
                {
                    this.RecycleOp.SetResult(TError);
                    if (this.RecycleOp.HasNotificationException)
                        return true;
                }
            }

            return false;
        }
        private Action<U> Fix<U>(Func<Action<U>, Action<U>> f)
        {
            return x => f(Fix(f))(x);
        }
        private Action<CancellationToken> CommonAct(Action<CancellationToken> Main)
        {
            var loop = this.Fix<CancellationToken>(f =>
            {
                f = token =>
                {
                    Task t_Success = Task.Run(() => Main(token), token).ContinueWith((t) =>
                    {
                        if (t.IsFaulted)
                        {
                            this.HasCancelled = false;
                            if (this.ConsumerFailEvent != null)
                                this.ConsumerFailEvent.Invoke(this);
                        }
                        if (t.IsCanceled)
                        {
                            this.HasCancelled = true;
                            if (this.ConsumerFailEvent != null)
                                this.ConsumerFailEvent.Invoke(this);
                        }
                        if (t.IsCompleted)
                        {
                            if (this.ConsumerFinishedEvent != null)
                                this.ConsumerFinishedEvent.Invoke(this);
                        }
                    });
                    t_Success.Wait();

                    if (this._cancellationTokenSource.IsCancellationRequested)
                    {
                        lock (this._lockObj)
                        {
                            this.isCompleted = true;
                            Monitor.PulseAll(this._lockObj);
                        }
                    }
                    else
                    {
                        Thread.Sleep(20);
                        if (this._messageCollection.Count < 1)
                        {
                            this._cancellationTokenSource.Cancel(false);
                            lock (this._lockObj)
                            {
                                this.isCompleted = true;
                                Monitor.PulseAll(this._lockObj);
                            }
                        }
                        else
                            f(token);
                    }
                };

                return f;
            });

            return loop;
        }
        private IEnumerable<T> GetConsuming(CancellationToken Token)
        {
            foreach (var message in this._messageCollection.GetConsumingEnumerable(Token))
            {
                if (this._messageCollection.Count > 0)
                    yield return message;
                else
                {
                    if (this._messageCollection.Count == 0 && message != null)
                    {
                        yield return message;
                        break;
                    }
                }
            }
        }

        #endregion
    }
}
